

use anyhow::bail;
use clap::{Parser, Subcommand};
use gfsc::*;
use std::{collections::HashMap};
use chrono;
use dirs::home_dir;
use time::{format_description, UtcOffset};
use tracing::error;
use tracing_subscriber::{
    self, filter::EnvFilter, fmt, fmt::time::OffsetTime, layer::SubscriberExt,
    util::SubscriberInitExt, Registry,
};
use std::path::{PathBuf,Path};
use serde::{Deserialize, Serialize};
use std::io::Write;
use std::fs::File;

type Result<T> = anyhow::Result<T>;



#[derive(Parser, Debug)]
#[clap(
    name = "parallel gfs",
    version,
    author,
    about = "parallel testing for gfsc"
)]
struct Cli {
    #[clap(subcommand)]
    command: Command,
    /// Sets a router.json
    #[clap(name = "config", short, long, default_value = "~/.rabbit/router.json")]
    router: String,
}

#[derive(Subcommand, Debug)]
enum Command {    
    ///  deploy a new operator
    Deploy {          
        #[clap(name = "fname", default_value = "")]
        fname: String,
    },    
    /// check router
    Router {   },    
    /// check if the operator is exist.
    Check {        
        #[clap(name = "fname", short, long, default_value = "")]
        fname: String,       
        #[clap(name = "ihash", short, long, default_value_t = 0)]
        ihash: i32,
        #[clap(name = "idx", short('x'), long, default_value_t = 0)]
        idx:i32,
        #[clap(name = "min", short('n'), long, default_value_t = 0)]
        min: i32,
        #[clap(name = "max", short, long, default_value_t = 255)]
        max:i32,
    },
    /// record 
    Record {        
        #[clap(name = "fname", short, long, default_value = "")]
        fname: String,       
        #[clap(name = "tkey", short, long, default_value = "")]
        tkey: String,       
        #[clap(name = "command", short, long, default_value = "append")]
        command: String,       
        #[clap(name = "ihash", short, long, default_value_t = 0)]
        ihash: i32,
        #[clap(name = "idx", short('x'), long, default_value_t = 0)]
        idx:i32,
        #[clap(name = "body", short, long, default_value = "this is a test")]
        body: String,       
    },
    /// 根据算子部署配置文件，执行新的计算。部署配置文件是部署命令生成，通常命名：算子-deploy.json
    Task { 
        /// 算子部署文件，如 ab-deploy.json, 内容来自deploy命令部署算子时生成的配置文件    
        #[clap(name = "fname", default_value = "ab-deploy.json")]
        fname: String,          
        /// 透明传递参数1
        #[clap(name = "arg1", short('a'), long, default_value = "arg1")]
        arg1: String,    
        /// 透明传递参数2       
        #[clap(name = "arg2", short('b'),long, default_value = "arg2")]
        arg2: String,    
        /// 透明传递参数3       
        #[clap(name = "arg3", short('c'),long, default_value = "arg3")]
        arg3: String,    
        /// 透明传递参数4       
        #[clap(name = "arg4", short('d'),long, default_value = "arg4")]
        arg4: String,    

       },    
}

#[derive(Serialize, Deserialize, Debug, Default)]
pub(crate) struct Deploy{
    name:String,
    writer:String,
    idx:i32,
    min:i32,
    max:i32
}

#[tokio::main]
pub async fn main() {

    let mut cli = Cli::parse();
    if cli.router.as_str().eq("~/.rabbit/router.json"){        
        let home =  home_dir().unwrap();
        cli.router = home.join(".rabbit").join("router.json").display().to_string();
    }
    tracelog_init();
    match cli.command {
      Command::Deploy {fname } => {        
        if let Err(err) = deploy_operator( &cli.router, &fname ).await{
            error!("{:#?}", err);
        }       
      },    
      Command::Check {fname , ihash, idx, min, max} => {
        if let Err(err) = check_operator(&cli.router, &fname, ihash, idx, min, max).await{
            println!("{:#?}", err);
        }
      },
      Command::Router{} => {
        println!("{:#?}", &cli.router);
        let router = GfsRouter::new(&cli.router) ;
        if let Err(err) = router.check() {
            println!("{:#?}", err);
        }else{
            println!("node-writer: {:#?}", router.get_dynasty_writers(router.cur_dynasty ) );
            println!("node-reader: {:#?}", router.get_dynasty_readers(router.cur_dynasty ) );
        }
      },       
      Command::Task {fname ,  arg1, arg2, arg3, arg4} => {
        /* let mut ids:HashMap<(i32,i32),i32> = HashMap::new();
        ids.insert((0, 127), idx);
        ids.insert((128, 255), idx); */        
        if let Err(err) = run_task(&cli.router, &fname, &arg1, &arg2, &arg3, &arg4).await{
            println!("{:#?}", err);
        }
      },

      Command::Record {fname , tkey, idx , ihash, command, body} => {        
        if let Err(err) = record_task(&cli.router, &fname, idx, ihash, &command, &tkey, body).await{
            println!("{:#?}", err);
        }
      },
  }

}
 

async fn deploy_operator(config:&str, fname:&str)->Result<()> {
    let exts:Vec<&str> = fname.split(".").collect();
    if exts.len()< 2 || !exts[exts.len()-1].eq("py") {
        println!("ext is error!");
        bail!("operator should be python script")
    }
    let router = GfsRouter::new(config);
    let ret = gfsc_operator_deploy(&router, fname).await?;    
    // let v = serde_json::to_value(&ret).unwrap();
    let writer = router.get_writer(router.cur_dynasty, 0)?;
    println!("{:#?}", ret);
    let mut rjson :Vec<Deploy> = vec![];
    for ((min, max), res) in ret.iter() {
        let p = Deploy{name:res.name.clone(), idx: res.idx, min:*min, max:*max, writer: writer.clone() };
        rjson.push(p);
    }
    
    let b = serde_json::to_string(&rjson).unwrap();
    let opfile = PathBuf::from(fname);
    let oname = opfile.file_name().unwrap().to_string_lossy();
    let oname:Vec<&str> = oname.split(".").collect();    

    let deploy_json = format!("{}-deploy.json", oname[0]);
    let mut buffer = File::create(&deploy_json)?;
    buffer.write_all(b.as_bytes())?;    
    println!("Result is {}",  &deploy_json);
    Ok(())   
}

async fn check_operator( config:&str, fname:&str, ihash:i32, idx:i32, min:i32, max:i32)->Result<()> {
    let router = GfsRouter::new(config);
    let mut operator_tables : HashMap<(i32, i32), GfsResponse> = HashMap::new();
    let  v =  GfsResponse{ name:fname.to_string(), ihash, idx, code:0};
    let ikey = (min, max);
    operator_tables.insert(ikey, v );
    
    if let Err(err)= gfsc_operator_exists(&router, &operator_tables).await{
        println!("{:#?}", err);
    }else{
        println!("config is ok!");
    }
    Ok(())   
}

async fn run_task( config:&str, fname:&str, arg1:&str, arg2:&str, arg3:&str, arg4:&str)->Result<()> {
    let router = GfsRouter::new(config);

    // 读取部署算子的信息文件，获取算子位置和writer
    let src = String::from_utf8(std::fs::read(fname)?)?;
    let dp: Vec<Deploy> = serde_json::from_str(src.as_str())?;
    println!("{:#?}", dp);

    // 构造符合计算的算子部署参数结构： HashMap < (min,max),idx >
    let op_name =dp[0].name.clone();
    let ihash = string_hash256(&op_name);
    let mut w_idx = 0;    
    let mut ids:HashMap<(i32,i32),i32> = HashMap::new();
    for i in 0..dp.len(){
        println!("node-{:#?}, operator is: ({}, {}, {}), range [{},{}]",  i, &op_name, ihash, dp[i].idx, dp[i].min, dp[i].max );    
        ids.insert((dp[i].min, dp[i].max),  dp[i].idx );
        if dp[i].min == 0 {
            w_idx = dp[i].idx
        }
    }
    // 构造传递给算子的参数结构
    let mut args:HashMap<String, String> = HashMap::new();
    args.insert("arg1".to_string(), arg1.to_string());
    args.insert("arg2".to_string(), arg2.to_string());    
    args.insert("arg3".to_string(), arg3.to_string());
    args.insert("arg4".to_string(), arg4.to_string());    

    args.insert(PARALLEL_RECORD_KEY.to_string(), dp[0].writer.clone());
    args.insert(PARALLEL_RECORD_IDX.to_string(), w_idx.to_string());

    // 重置算子的计算状态
    gfsc_record_clear(&router, ihash, 7, &op_name).await?;
    // 启动新的运算， 具体执行情况可以访问提供给算子的writer服务的/parallel/status/operator_name了解
    gfsc_map_task(&router, &op_name, ids, &mut args).await     
}


async fn record_task( config:&str, fname:&str, idx:i32, ihash: i32, command: &str, tkey:&str,  body: String)->Result<()> {
    let router = GfsRouter::new(config);
    gfsc_record(&router, ihash, idx, fname, command, tkey, -1, bytes::Bytes::from(body)).await  
}

fn tracelog_init() {
    let secs = chrono::Local::now().offset().local_minus_utc();
    let offset = UtcOffset::from_whole_seconds(secs).unwrap();
    // 同时输出到文件和stderr
    let env_filter =
        EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("debug"));    
    let format = "[year]-[month]-[day] [hour]:[minute]:[second].[subsecond digits:3]";    
    // 输出到控制台 stderr,
    
    let formatting_layer =       
            fmt::layer()
            .pretty()               
            .with_writer(std::io::stderr)
            .with_timer(OffsetTime::new(
                offset,
                format_description::parse(format).unwrap(),
            ));
// 注册
        Registry::default()
            .with(env_filter)
            .with(formatting_layer)
            .init();        
}